package com.hyq.learn.rabbitmq.consumer;

import com.hyq.learn.common.util.SnakeJsonUtils;
import com.hyq.learn.rabbitmq.constant.ExchangeConstants;
import com.hyq.learn.rabbitmq.constant.QueueConstants;
import com.hyq.learn.rabbitmq.constant.RoutingKeyConstants;
import com.hyq.learn.rabbitmq.handle.RabbitMessageHandle;
import com.hyq.learn.rabbitmq.message.CommonMessage;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

/**
 * @author heyuqin
 * @date 2024/8/20 17:20
 */
@Component
public class RabbitConsumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitConsumer.class);

    @Autowired
    private RabbitMessageHandle handle;

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "test"), exchange = @Exchange(value = "test")))
    public void simpleConsumer(Message message, Channel channel) {
        LOGGER.info("receive message : {}", message);
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QueueConstants.DIRECT_QUEUE),
            exchange = @Exchange(value = ExchangeConstants.DIRECT_EXCHANGE),
            key = RoutingKeyConstants.DIRECT_ROUTING_KEY
    ))
    public void directConsumer1(Message message, Channel channel) {
        LOGGER.info("direct consumer receive message,queue:{},exchange:{},message:{}",
                QueueConstants.DIRECT_QUEUE, ExchangeConstants.DIRECT_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    // fanout 广播类型消费者

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QueueConstants.FANOUT_QUEUE_1),
            exchange = @Exchange(value = ExchangeConstants.FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT)))
    public void fanoutConsumer1(Message message, Channel channel) {
        LOGGER.info("fanout consumer_1 receive message,queue:{},exchange:{},message:{}",
                QueueConstants.FANOUT_QUEUE_1, ExchangeConstants.FANOUT_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    /**
     * 为实现，一条消息被多个消费者消费，不指定队列名称，广播模式只需要交换机即可，
     * 本质上是将生成队列的过程交给了mq去实现，和广播模型并没有任何区别，当项目停止会自动删除对应的队列。
     *
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue(),
            exchange = @Exchange(value = ExchangeConstants.FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT)))
    public void fanoutConsumer1_2(Message message, Channel channel) {
        LOGGER.info("fanout consumer_1_2 receive message,queue:{},exchange:{},message:{}",
                QueueConstants.FANOUT_QUEUE_1, ExchangeConstants.FANOUT_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QueueConstants.FANOUT_QUEUE_2),
            exchange = @Exchange(value = ExchangeConstants.FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT)))
    public void fanoutConsumer2(Message message, Channel channel) {
        LOGGER.info("fanout consumer_2 receive message,queue:{},exchange:{},message:{}",
                QueueConstants.FANOUT_QUEUE_2, ExchangeConstants.FANOUT_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QueueConstants.FANOUT_QUEUE_3),
            exchange = @Exchange(value = ExchangeConstants.FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT)))
    public void fanoutConsumer3(Message message, Channel channel) {
        LOGGER.info("fanout consumer_3 receive message,queue:{},exchange:{},message:{}",
                QueueConstants.FANOUT_QUEUE_3, ExchangeConstants.FANOUT_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    // topic 类型消费者

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QueueConstants.TOPIC_QUEUE_1),
            exchange = @Exchange(value = ExchangeConstants.TOPIC_EXCHANGE, type = ExchangeTypes.TOPIC),
            key = RoutingKeyConstants.TOPIC_ROUTING_KEY_1))
    public void topicConsumer1(Message message, Channel channel) {
        LOGGER.info("topic consumer_1 receive message,queue:{},exchange:{},message:{}",
                QueueConstants.TOPIC_QUEUE_1, ExchangeConstants.TOPIC_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QueueConstants.TOPIC_QUEUE_2),
            exchange = @Exchange(value = ExchangeConstants.TOPIC_EXCHANGE, type = ExchangeTypes.TOPIC),
            key = RoutingKeyConstants.TOPIC_ROUTING_KEY_2))
    public void topicConsumer2(Message message, Channel channel) {
        LOGGER.info("topic consumer_2 receive message,queue:{},exchange:{},message:{}",
                QueueConstants.TOPIC_QUEUE_2, ExchangeConstants.TOPIC_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }

    // 死信队列 消费者
    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QueueConstants.DEAD_LETTER_QUEUE_PROCESS, durable = "true"),
            exchange = @Exchange(value = ExchangeConstants.DEAD_LETTER_EXCHANGE, type = ExchangeTypes.FANOUT)))
    public void deadLetterConsumer(Message message, Channel channel) {
        LOGGER.info("deadLetter consumer receive message,queue:{},exchange:{},message:{}",
                QueueConstants.DEAD_LETTER_QUEUE_PROCESS, ExchangeConstants.DEAD_LETTER_EXCHANGE, new String(message.getBody()));
        Consumer<CommonMessage> consumer = commonMessage -> {
            // 处理消息
            System.out.println("消费消息：" + SnakeJsonUtils.toJson(commonMessage));
        };
        handle.handle(message, channel, CommonMessage.class, consumer);
    }
}
